package app.dws;

import app.dwd.BaseAppFollowV2;

import static common.Constant.*;

import bean.OrderWide;
import bean.PaymentWide;
import bean.ProductStats;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import common.Constant;
import function.DIMAsyncFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import redis.clients.jedis.Jedis;
import util.DimUtil;
import util.GmallSinkUtil;
import util.MyCommonUtils;

import java.sql.Connection;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

public class DWSProductStatsApp extends BaseAppFollowV2 {

    public static void main(String[] args) {
        new DWSProductStatsApp().init(4234,
                                      1,
                                      "DWSProductStatsApp",
                                      "DWSProductStatsApp",
                                      TOPIC_DWD_PAGE_LOG,
                                      TOPIC_DWD_DISPLAY_LOG,
                                      TOPIC_DWD_FAVOR_INFO,
                                      TOPIC_DWD_CART_INFO,
                                      TOPIC_DWM_ORDER_WIDE,
                                      TOPIC_DWM_PAYMENT_WIDE,
                                      TOPIC_DWD_ORDER_REFUND_INFO,
                                      TOPIC_DWD_COMMENT_INFO);
    }

    @Override
    public void run(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streamKafkaSourcehashMap) {
        //1. pojo封装,合并各流
        DataStream<ProductStats> productStatsDataStream = parseAndUnion(streamKafkaSourcehashMap);
        //2. sku_id分组,水印,开窗,聚合
        SingleOutputStreamOperator<ProductStats> aggSkuIdStats = aggregateByWindowAndSkuId(productStatsDataStream);
        //3.异步查询维度进行关联
        SingleOutputStreamOperator<ProductStats> joinDimProductStats = joinDim(aggSkuIdStats);
        //4.写入ck
        joinDimProductStats.addSink(GmallSinkUtil.getClickHouseSink(CLICKHOUSE_DATABASE, CLICKHOUSEPRODUCTSTATS2021, ProductStats.class));
        //5.写入kafka,供商品行为关键词主题宽表使用
        GmallSinkUtil.sendToKafka(joinDimProductStats.map(JSON::toJSONString), TOPIC_DWS_PRODUCT_STATS_2021);


    }


    private SingleOutputStreamOperator<ProductStats> joinDim(SingleOutputStreamOperator<ProductStats> aggSkuIdStats) {
        return AsyncDataStream.unorderedWait(aggSkuIdStats,
                                             new DIMAsyncFunction<ProductStats>() {
                                                 @Override
                                                 public void addDim(Connection conn,
                                                                    Jedis jedis,
                                                                    ProductStats input,
                                                                    ResultFuture<ProductStats> resultFuture) throws Exception {
                                                     JSONObject skuInfo = DimUtil.readDim(conn, jedis, DIM_SKU_INFO, input.getSku_id(), 1);
                                                     input.setSku_name(skuInfo.getString("SKU_NAME"));
                                                     input.setSku_price(skuInfo.getBigDecimal("PRICE"));
                                                     input.setSpu_id(skuInfo.getLong("SPU_ID"));
                                                     input.setTm_id(skuInfo.getLong("TM_ID"));
                                                     input.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));

                                                     JSONObject spuInfo = DimUtil.readDim(conn, jedis, DIM_SPU_INFO, input.getSku_id(), 1);
                                                     input.setSpu_name(spuInfo.getString("SPU_NAME"));

                                                     JSONObject baseTrademark = DimUtil.readDim(conn, jedis, DIM_BASE_TRADEMARK, input.getSku_id(), 1);
                                                     input.setTm_name(baseTrademark.getString("TM_NAME"));

                                                     JSONObject baseCategory3 = DimUtil.readDim(conn, jedis, DIM_BASE_CATEGORY3, input.getSku_id(), 1);
                                                     input.setCategory3_name(baseCategory3.getString("NAME"));
                                                     resultFuture.complete(Collections.singleton(input));
                                                 }
                                             },
                                             300,
                                             TimeUnit.SECONDS);
    }

    private SingleOutputStreamOperator<ProductStats> aggregateByWindowAndSkuId(DataStream<ProductStats> productStatsDataStream) {
        SingleOutputStreamOperator<ProductStats> aggSkuStatsStream = productStatsDataStream
            .assignTimestampsAndWatermarks(WatermarkStrategy
                                               .<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(15))
                                               .withTimestampAssigner((obj, ts) -> obj.getTs()))
            .keyBy(k -> k.getSku_id())
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .sideOutputLateData(new OutputTag<ProductStats>("late") {
            })
            .reduce(new ReduceFunction<ProductStats>() {
                @Override
                public ProductStats reduce(ProductStats value1, ProductStats value2) throws Exception {
                    value1.setRefund_amount(value1.getRefund_amount().add(value2.getRefund_amount()));
                    value1.setPayment_amount(value1.getPayment_amount().add(value2.getPayment_amount()));
                    value1.setOrder_amount(value1.getOrder_amount().add(value2.getOrder_amount()));
                    value1.setOrder_sku_num(value1.getOrder_sku_num() + value2.getOrder_sku_num());
                    value1.setCart_ct(value1.getCart_ct() + value2.getCart_ct());
                    value1.setFavor_ct(value1.getFavor_ct() + value2.getFavor_ct());
                    value1.setDisplay_ct(value1.getDisplay_ct() + value2.getDisplay_ct());
                    value1.setComment_ct(value1.getComment_ct() + value2.getComment_ct());
                    value1.setGood_comment_ct(value1.getGood_comment_ct() + value2.getGood_comment_ct());
                    value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                    value1.getRefundOrderIdSet().addAll(value2.getRefundOrderIdSet());
                    value1.getPaidOrderIdSet().addAll(value2.getPaidOrderIdSet());
                    value1.setClick_ct(value1.getClick_ct() + value2.getClick_ct());

                    return value1;
                }
            }, new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                @Override
                public void process(Long aLong, Context ctx, Iterable<ProductStats> elements, Collector<ProductStats> out) throws Exception {
                    ProductStats productStats = elements.iterator().next();
                    productStats.setOrder_ct((long) productStats.getOrderIdSet().size());
                    productStats.setPaid_order_ct((long) productStats.getPaidOrderIdSet().size());
                    productStats.setRefund_order_ct((long) productStats.getRefundOrderIdSet().size());
                    long start = ctx.window().getStart();
                    long end = ctx.window().getEnd();
                    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    productStats.setStt(simpleDateFormat.format(start));
                    productStats.setEdt(simpleDateFormat.format(end));
                    out.collect(productStats);
                }
            });
        return aggSkuStatsStream;


    }

    private DataStream<ProductStats> parseAndUnion(HashMap<String, DataStreamSource<String>> dsMap) {
        SingleOutputStreamOperator<ProductStats> pageStats = dsMap
            .get(TOPIC_DWD_PAGE_LOG)
            .flatMap(new FlatMapFunction<String, ProductStats>() {
                @Override
                public void flatMap(String value, Collector<ProductStats> out) throws Exception {
                    JSONObject obj = JSON.parseObject(value);
                    JSONObject page = obj.getJSONObject("page");
                    if ("good_detail".equals(page.getString("page_id"))
                        && "sku_id".equals(page.getString("item_type"))) {
                        ProductStats productStats = new ProductStats();

                        productStats.setSku_id(page.getLong("item"));
                        productStats.setClick_ct(1L);
                        productStats.setTs(obj.getLong("ts"));
                        out.collect(productStats);
                    }
                }
            });

        SingleOutputStreamOperator<ProductStats> displayStats = dsMap
            .get(TOPIC_DWD_DISPLAY_LOG)
            .flatMap(new FlatMapFunction<String, ProductStats>() {
                @Override
                public void flatMap(String value, Collector<ProductStats> out) throws Exception {
                    JSONObject obj = JSON.parseObject(value);
                    if ("sku_id".equals(obj.getString("item_type"))) {
                        ProductStats productStats = new ProductStats();

                        productStats.setSku_id(obj.getLong("item"));
                        productStats.setDisplay_ct(1L);
                        productStats.setTs(obj.getLong("ts"));

                        out.collect(productStats);
                    }
                }
            });

        SingleOutputStreamOperator<ProductStats> favorStats = dsMap
            .get(TOPIC_DWD_FAVOR_INFO)
            .map(w -> {
                JSONObject obj = JSON.parseObject(w);

                ProductStats productStats = new ProductStats();
                productStats.setSku_id(obj.getLong("sku_id"));
                productStats.setFavor_ct(1L);
                productStats.setTs(MyCommonUtils.toLongTypeTs(obj.getString("create_time")));

                return productStats;
            });

        SingleOutputStreamOperator<ProductStats> cartStats = dsMap
            .get(TOPIC_DWD_CART_INFO)
            .map(w -> {
                JSONObject obj = JSON.parseObject(w);

                ProductStats productStats = new ProductStats();
                productStats.setSku_id(obj.getLong("sku_id"));
                productStats.setCart_ct(1L);
                productStats.setTs(MyCommonUtils.toLongTypeTs(obj.getString("create_time")));

                return productStats;
            });

        SingleOutputStreamOperator<ProductStats> orderWideStream = dsMap
            .get(TOPIC_DWM_ORDER_WIDE)
            .map(w -> {
                OrderWide orderWide = JSON.parseObject(w, OrderWide.class);

                ProductStats productStats = new ProductStats();
                productStats.setSku_id(orderWide.getSku_id());
                productStats.getOrderIdSet().add(orderWide.getOrder_id());

                productStats.setTs(MyCommonUtils.toLongTypeTs(orderWide.getCreate_time()));
                productStats.setOrder_sku_num(orderWide.getSku_num());
                productStats.setOrder_amount(orderWide.getSplit_total_amount());

                return productStats;
            });

        SingleOutputStreamOperator<ProductStats> paymentStats = dsMap
            .get(TOPIC_DWM_PAYMENT_WIDE)
            .map(w -> {
                PaymentWide paymentWide = JSON.parseObject(w, PaymentWide.class);

                ProductStats productStats = new ProductStats();
                productStats.setPayment_amount(paymentWide.getSplit_total_amount());
                productStats.getPaidOrderIdSet().add(paymentWide.getOrder_id());
                productStats.setSku_id(paymentWide.getSku_id());
                productStats.setTs(MyCommonUtils.toLongTypeTs(paymentWide.getPayment_create_time()));

                return productStats;
            });

        SingleOutputStreamOperator<ProductStats> refundStats = dsMap
            .get(TOPIC_DWD_ORDER_REFUND_INFO)
            .map(w -> {
                JSONObject obj = JSON.parseObject(w);

                ProductStats productStats = new ProductStats();
                productStats.setSku_id(obj.getLong("sku_id"));
                productStats.setRefund_amount(obj.getBigDecimal("refund_amount"));
                productStats.getRefundOrderIdSet().add(obj.getLong("order_id"));
                productStats.setTs(MyCommonUtils.toLongTypeTs(obj.getString("create_time")));
                return productStats;
            });

        SingleOutputStreamOperator<ProductStats> commentStats = dsMap
            .get(TOPIC_DWD_COMMENT_INFO)
            .map(w -> {
                JSONObject obj = JSON.parseObject(w);

                ProductStats productStats = new ProductStats();
                productStats.setSku_id(obj.getLong("sku_id"));
                productStats.setTs(MyCommonUtils.toLongTypeTs(obj.getString("create_time")));
                productStats.setComment_ct(1L);

                if (FIVE_START_GOOD_COMMENT.equals(obj.getString("appraise"))
                    || FOUR_START_GOOD_COMMENT.equals(obj.getString("appraise"))) {
                    productStats.setGood_comment_ct(1L);
                }
                return productStats;
            });
        DataStream<ProductStats> productStatsDataStream = pageStats.union(commentStats,
                                                                          refundStats,
                                                                          paymentStats,
                                                                          orderWideStream,
                                                                          cartStats,
                                                                          favorStats,
                                                                          displayStats);
        return productStatsDataStream;

    }
}
